package cluster

import (
	"context"
	"sync"
	"time"

	"gitee.com/zfd81/dbrs/internal/token"
	pb "gitee.com/zfd81/dbrs/proto/dbrspb"
	"google.golang.org/grpc"

	"gitee.com/zfd81/dbrs/sys"

	"github.com/spf13/cast"

	"gitee.com/zfd81/dbrs/cluster/msgpb"

	"github.com/sirupsen/logrus"
)

type delegate struct {
	mu sync.RWMutex
	id string
}

// NodeMeta 用于在广播活动消息时检索有关当前节点的元数据。
// 它的长度限制为给定的字节大小。此元数据在节点结构中可用。
func (d *delegate) NodeMeta(limit int) []byte {
	return []byte{}
}

// NotifyMsg 收到用户数据消息时调用NotifyMsg。
// 应注意此方法不会阻塞，因为这样做会阻塞整个UDP数据包接收循环。
// 此外，可以在调用返回后修改字节片，因此如果需要，应该复制它
func (d *delegate) NotifyMsg(bytes []byte) {
	length := len(bytes)
	if length == 0 {
		return
	}
	msg, err := UnmarshalMessage(bytes)
	if err != nil {
		logrus.Infof("cluster notify msg error: %s", err)
	}

	id := msg.Id()
	d.mu.Lock()
	if id == d.id {
		d.mu.Unlock()
		return
	}
	d.id = id
	d.mu.Unlock()

	switch msg.Action() {
	case msgpb.Action_ELECT:
		id := msg.Get(0)
		if leader.Id != id {
			flag := leader.Id == ""
			mtx.Lock()
			leader.Id = id
			leader.TimeStamp = cast.ToInt64(msg.Get(1))
			logrus.Infof("%s elected leader", LeaderMember().ServAddress())
			if flag {
				syncUsers()
			}
			mtx.Unlock()
		}
	case msgpb.Action_ADD_USER:
		user := sys.NewUser(msg.Get(0), msg.Get(1))
		err = user.Save()
		if err != nil {
			logrus.Errorf("Sync add user '%s' error: %v\n", user.Name, err)
			return
		}
		logrus.Infof("Sync add user '%s' successfully.", user.Name)
	case msgpb.Action_PWD_USER:
		user := &sys.User{
			Name:     msg.Get(0),
			Password: msg.Get(1),
		}
		err = user.Modify()
		if err != nil {
			logrus.Errorf("Sync modify user '%s' password error: %v\n", user.Name, err)
			return
		}
		logrus.Infof("Sync modify user '%s' password successfully.", user.Name)
	case msgpb.Action_DEL_USER:
		err = sys.DeleteUser(msg.Get(0))
		if err != nil {
			logrus.Errorf("Sync delete user '%s' error: %v\n", msg.Get(0), err)
			return
		}
		logrus.Infof("Sync delete user '%s' successfully.", msg.Get(0))
	case msgpb.Action_SIGN_USER:
		user := &sys.User{
			Name:             msg.Get(0),
			DefaultSignature: msg.Get(1),
		}
		err = user.Modify()
		if err != nil {
			logrus.Errorf("Sync set user '%s' signature error: %v\n", user.Name, err)
			return
		}
		logrus.Infof("Sync set user '%s' signature successfully.", user.Name)
	}

}

// GetBroadcasts 当可以广播用户数据消息时，调用GetBroadcasts。
// 它可以返回要发送的缓冲区列表。每个缓冲区应承担一定的开销，并对允许的总字节大小进行限制。
// 要发送的结果数据的总字节大小不得超过限制。
// 应注意此方法不会阻塞，因为这样做会阻塞整个UDP数据包接收循环。
func (d *delegate) GetBroadcasts(overhead, limit int) [][]byte {
	return broadcasts.GetBroadcasts(overhead, limit)
}

// LocalState 用于TCP推/拉。
// 除了成员信息之外，还会将其发送到远程端。任何数据都可以发送到这里。
// 请参见MergeRemoteState。“join”布尔值表示这是用于连接，而不是推/拉。
// 每隔PushPullInterval周期，本地memberlist回调LocalState方法，把本地全部数据发送到其他节点；
// 其他节点memberlist回调MergeRemoteState，接收数据进行同步。
func (d *delegate) LocalState(join bool) []byte {
	str := Self().ServAddress() + " [" + time.Now().Format("2006-01-02 15:04:05") + "]"
	logrus.Debugf("LocalState: %v %s \n", join, str)
	return []byte(str)
}

// MergeRemoteState 在TCP推/拉之后调用MergeRemoteState。
// 这是从远程端接收到的状态，是远程端LocalState调用的结果。“join”布尔值表示这是用于连接，而不是推/拉。
func (d *delegate) MergeRemoteState(buf []byte, join bool) {
	if len(buf) == 0 {
		return
	}
	logrus.Debugf("MergeRemoteState: %v %s \n", join, string(buf))
}

func syncUsers() {
	//同步系统用户
	claims := token.BrokerClaims{
		Tenant: "ZFD20220726",
		Sign:   "ZFD20220726",
		User:   "ZFD20220726",
	}
	tokenstr, err := token.Create(claims, "zfd1q2w3e", 1*60)
	if err != nil {
		logrus.Fatalf("Cannot synchronize user information with leader, %v\n", err)
	}
	request := &pb.FindAllUsersRequest{
		Token:  tokenstr,
		Prefix: false,
	}

	leader := LeaderMember()
	conn, err := grpc.Dial(leader.CtlAddress(), grpc.WithInsecure(), grpc.WithBlock(), grpc.WithTimeout(3*time.Second))
	if err != nil {
		logrus.Fatalf("Cannot synchronize user information with leader, %v\n", err)
	}
	resp, err := pb.NewBrokerSystemClient(conn).FindAllUsers(context.Background(), request)
	if err != nil {
		logrus.Fatalf("Cannot synchronize user information with leader, %v\n", err)
	}

	users := []*sys.User{}
	for _, u := range resp.Users {
		ctime, _ := time.ParseInLocation("2006-01-02 15:04:05", u.CreateTime, time.Local)
		mtime, _ := time.ParseInLocation("2006-01-02 15:04:05", u.ModifyTime, time.Local)
		users = append(users, &sys.User{
			Name:             u.Name,
			Password:         u.Password,
			DefaultSignature: u.Sign,
			CreateTime:       ctime,
			ModifyTime:       mtime,
		})
	}

	err = sys.TruncateUser()
	if err != nil {
		logrus.Fatalf("Cannot synchronize user information with leader, %v\n", err)
	}

	err = sys.BatchAddUsers(users)
	if err != nil {
		logrus.Fatalf("Cannot synchronize user information with leader, %v\n", err)
	}
}
